package com.atguigu.flink.datastreamapi.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/6
 *  加入 kafka-connector
 */
public class Demo2_KafkaSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //新的sinkAPI，需要开启ck
        //env.enableCheckpointing(5000);

        //准备KafkaSink  角色：生产者
        KafkaSink<String> kafkaSink = KafkaSink
            .<String>builder()
            .setBootstrapServers("hadoop102:9092")
            //设置序列化器
            .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                                              .setTopic("topicC") //写到哪个主题
                                              .setValueSerializationSchema(new SimpleStringSchema()) //value的反序列化器必须要设置
                                              .build()
                //.setPartitioner() //指定自定义的分区器，不指定，默认
            )
            //其他的生产者设置
            //.setProperty()
            //语义保证  EOS需要开启CK
            .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
            //设置kafka生产者事务id的前缀，可以在broker中的_transaction_state中查看。EOS，必须要设置
            .setTransactionalIdPrefix("atguigu-")
            .build();

         env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .map(JSON::toJSONString)
            .sinkTo(kafkaSink);



        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
